package com.duobaoyu.dby.supply.message.listener;

import com.alibaba.fastjson.JSON;
import com.duobaoyu.dby.common.core.constant.RequestConstants;
import com.duobaoyu.dby.resource.remote.common.constant.ResourceMessageConstant;
import com.duobaoyu.dby.supply.common.mq.remote.message.base.BaseMessage;
import com.duobaoyu.dby.supply.core.rocketmq.listener.AbstractClusterListener;
import com.duobaoyu.dby.supply.message.constant.ConsumerGroupConstant;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.slf4j.MDC;
import org.springframework.stereotype.Component;

/**
 * 异步任务消费消息监听
 * @Author yangyong
 * @Date 2023/10/20
 */
@Slf4j
@Component
@RequiredArgsConstructor
@RocketMQMessageListener(topic = ResourceMessageConstant.AsyncTaskTopic.TOPIC, selectorExpression = ResourceMessageConstant.AsyncTaskTopic.NEW_SUPPLY_CHAIN_TAG, consumerGroup = ConsumerGroupConstant.ASYNC_TASK)
public class SupplyChainAsyncTaskConsumer extends AbstractClusterListener{
	@Override
	public boolean ifRetry(MessageExt message,Exception exception) { return Boolean.TRUE; }

	@Override
	protected void beforeHandler(MessageExt message){
		try {
			BaseMessage baseMessage = JSON.parseObject(new String(message.getBody()), BaseMessage.class);
			MDC.put(RequestConstants.REQUEST_ID, baseMessage.getTraceId());
		}catch (Exception e){
			MDC.put(RequestConstants.REQUEST_ID, message.getMsgId());
		}
		if(log.isDebugEnabled()) {
			log.debug("异步任务消息Consumer接收到消息，messageId={},message={},topic={},tag={}", message.getMsgId(), message, message.getTopic(), message.getTags());
		}
	}

	@Override
	protected void finallyHandler(MessageExt message){
		MDC.remove(RequestConstants.REQUEST_ID);
		if(log.isDebugEnabled()) {
			log.debug("异步任务消息Consumer接收到消息，message={}", message);
		}
	}
}
